#include "message_bus.h"
#include "timer.hpp"
#include <memory>
#include <chrono>
#include <algorithm>


static uint64_t getTimeStamp()
{
    std::chrono::microseconds ms = std::chrono::duration_cast<std::chrono::microseconds>(
        std::chrono::system_clock::now().time_since_epoch()
        );

    return ms.count();
}

bool MessageBus::subscribe(int msg, CallbackItem_ptr item)
{
    std::unique_lock<std::mutex> lck(_callbackMapMutex);
    if (_callbackMap.find(msg) != _callbackMap.end())
    {
        auto& itemVec = _callbackMap[msg];
        if (find(itemVec.begin(), itemVec.end(), item) == itemVec.end())
            itemVec.push_back(item);
    }
    else
    {
        std::vector<CallbackItem_ptr> newItemVec;
        newItemVec.push_back(item);
        _callbackMap.insert(CallbackMap_t::value_type(msg, newItemVec));
    }

    return true;
}

bool MessageBus::subscribe(CallbackItem_t src)
{
    auto& item = std::make_shared<CallbackItem_t>(src);
    if (item->msgNumVec.empty())
        return false;

    std::unique_lock<std::mutex> lck(_timeoutCheckListMutex);
    for (auto& tmp : _timeoutCheckList)
    {
        if (tmp->msgNumVec == item->msgNumVec
            && tmp->callback.target<Callback_t>() == item->callback.target<Callback_t>()
            && tmp->timeOutCallback.target<TimeOutCallback_t>() == item->timeOutCallback.target<TimeOutCallback_t>())
        {
            return false;	// Already exist,but timeoutInterval may not equal.
        }
    }
    lck.unlock();

    for (auto& m : item->msgNumVec)
    {
        if (!subscribe(m, item))
            return false;
    }

    if (item->timeOutCallback != nullptr)
        regTimeOutCallback(item);

    return true;
}

bool MessageBus::unsubscribe(int msg, CallbackItem_ptr item)
{
    std::unique_lock<std::mutex> lck(_callbackMapMutex);
    if (_callbackMap.find(msg) != _callbackMap.end())
    {
        auto& callbackItemVec= _callbackMap[msg];
        auto& result = find(callbackItemVec.begin(), callbackItemVec.end(), item);
        if (result != callbackItemVec.end())
        {
            callbackItemVec.erase(result);
            return true;
        }
    }
    return false;
}

void MessageBus::regTimeOutCallback(CallbackItem_ptr item)
{
    item->timeoutStamp = item->timeoutInterval*1000 + getTimeStamp();

    std::unique_lock<std::mutex> lck(_timeoutCheckListMutex);

    auto it = std::find_if(_timeoutCheckList.begin(), _timeoutCheckList.end(), [item](CallbackItem_ptr tmp) {
        return (tmp->timeoutStamp > item->timeoutStamp); });
    if (it != _timeoutCheckList.end())
        _timeoutCheckList.insert(it, item);
    else
        _timeoutCheckList.push_back(item);
}

void MessageBus::start()
{
    if (_timer.is_stopped())
    {
        _timer.start(100, std::bind(&MessageBus::timeOutCheck, this));
    }
}

void MessageBus::publish(int msg, std::string param1, int param2)
{
    std::unique_lock<std::mutex> lck1(_timeoutCheckListMutex);

    // Erase the relational items of the message from the _timeoutCheckList.
    for (auto& it = _timeoutCheckList.begin(); it != _timeoutCheckList.end(); ++it)
    {
        // not normal, try muti
        auto& item = *it;
        if (find(item->msgNumVec.begin(), item->msgNumVec.end(), msg)
            != item->msgNumVec.end())
        {
            _timeoutCheckList.erase(it);
            break;
        }
    }
    lck1.unlock();

    std::unique_lock<std::mutex> lck2(_callbackMapMutex);
    // Call back. and erase the message from callbacklist if the callback vector of the message is empty.
    if (_callbackMap.find(msg) == _callbackMap.end())
        return;

    for (auto& it = _callbackMap[msg].begin(); it != _callbackMap[msg].end();)
    {
        auto& item = *it;

        if (item->callback != nullptr)
            item->callback(param1, param2);

        if (item->callbackType == CallbackType_t::ALWAYS)
        {
            ++it;
            continue;
        }
           
        // Since one of the messages in item->msgNumVec is come,
        // we should erase other messages in item->msgNumVec
        for(auto m : item->msgNumVec)
        {
            if (_callbackMap.find(m) != _callbackMap.end() && msg != m)
            {
                auto& itemVec = _callbackMap[m];
                auto& result = find(itemVec.begin(), itemVec.end(), item);
                if (result != itemVec.end())
                {
                    itemVec.erase(result);

                    // erase the message from callbacklist if the callback vector of the message is empty.
                    if (itemVec.empty())
                        _callbackMap.erase(m);
                    continue;
                }
            }
        }
        it = _callbackMap[msg].erase(it);
    }
    // erase the message from callbacklist if the callback vector of the message is empty.
    if (_callbackMap[msg].empty())
        _callbackMap.erase(msg);
    lck2.unlock();
}

void MessageBus::timeOutCheck()
{
    std::unique_lock<std::mutex> lck(_timeoutCheckListMutex);

    uint64_t currentTimeStamp = getTimeStamp();
    for (auto& it =_timeoutCheckList.begin(); it != _timeoutCheckList.end();)
    {
        if ((*it)->timeoutStamp > currentTimeStamp) // not timeout
        {
            ++it;
            continue;
        }

        if ((*it)->timeOutCallback != nullptr)
            (*it)->timeOutCallback();

        for (auto m : (*it)->msgNumVec)
            unsubscribe(m, *it);

        it = _timeoutCheckList.erase(it);
    }
}

void MessageBus::reset()
{
    std::unique_lock<std::mutex> lck1(_callbackMapMutex);
    std::unique_lock<std::mutex> lck2(_timeoutCheckListMutex);

    _timeoutCheckList.clear();
    _callbackMap.clear();
}

void MessageBus::stop()
{
    _timer.stop();
}